package cn.doitedu.dw.etl

import java.net.URI
import java.util
import java.util.{Properties, UUID}

import ch.hsr.geohash.GeoHash
import cn.doitedu.dw.utils.BeanUtil
import org.apache.commons.lang3.StringUtils
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.{FSDataInputStream, FileSystem, Path}
import org.apache.hadoop.util.bloom.{BloomFilter, Key}
import org.apache.hadoop.util.hash.Hash
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.types.{DataTypes, StructType}
import org.lionsoul.ip2region.{DbConfig, DbSearcher}
import redis.clients.jedis.{GeoRadiusResponse, GeoUnit, Jedis}

/**
 * 将APP日志的ODS层表，计算加工成 DWD 表数据
 *
 */
object OdsAppLog2DwdApp_BloomFilter {

  def main(args: Array[String]): Unit = {

    if (args.size < 3) {
      println(
        """
          |usage:
          |  args(0): 运行模式
          |  args(1): 评分表的T-1日期
          |  args(2): 日志表的T日期
          |
          |""".stripMargin)
      sys.exit(1)
    }


    /**
     * 构造一个sparksession实例
     * 开启hive支持
     * （ hive的支持，需要enableHiveSupport()，还要在resources中放入hive-site.xml,core-site.xml配置文件
     * 还要在pom.xml中引入spark-hive的依赖
     * ）
     */
    val spark = SparkSession.builder()
      .config("spark.sql.shuffle.partitions", "2")
      // 设置spark的序列化器为 ： Kryo序列化器
      .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
      .appName(OdsAppLog2DwdApp.getClass.getSimpleName)
      .enableHiveSupport()
      .master(args(0))
      .getOrCreate()

    // 导入隐式转换
    import spark.implicits._

    /*spark.sql(
      """
        |
        |show tables from ods18
        |
        |""".stripMargin).show(100,false)*/


    // 利用hive功能去读、解析，hive中表
    val odsLog = spark.read.table("ods18.event_app_log").where(s"dt='${args(2)}'")

    /* 自己直接读取hive表的hdfs上的文件，来解析
       val schema = new StructType()
      .add("account", DataTypes.StringType)
      .add("appId", DataTypes.StringType)
      .add("appVersion", DataTypes.StringType)
      .add("carrier", DataTypes.StringType)
      .add("deviceId", DataTypes.StringType)
      .add("deviceType", DataTypes.StringType)
      .add("eventId", DataTypes.StringType)
      .add("ip", DataTypes.StringType)
      .add("latitude", DataTypes.DoubleType)
      .add("longitude", DataTypes.DoubleType)
      .add("netType", DataTypes.StringType)
      .add("osName", DataTypes.StringType)
      .add("osVersion", DataTypes.StringType)
      .add("properties", DataTypes.createMapType(DataTypes.StringType,DataTypes.StringType))
      .add("releaseChannel", DataTypes.StringType)
      .add("resolution", DataTypes.StringType)
      .add("sessionId", DataTypes.StringType)
      .add("timeStamp", DataTypes.LongType)
    val df = spark.read.schema(schema).json("/user/hive/warehouse/ods18.db/event_app_log/dt=2020-12-11/").limit(10)*/

    /**
     * 过滤
     */
    val filtered = odsLog
      // nvl(a,b)返回a，b中不为null的值，如果都为null，则返回null
      .where("nvl(account,deviceid) is not null")
      // 过滤掉 properties为空，或eventid为空，或sessionid为空的记录
      .where("(properties is not null) and  (eventid is not null and trim(eventid) != '') and (sessionid is not null and trim(sessionid) != '') ")
    // .where("过滤掉时间范围不符合要求记录") TODO


    /**
     * session分割
     */

    // 把dataframe转成CaseClass的RDD
    /**
     * 数据规范化
     * 比如，空串，统一成Null
     * 比如，boolean类型字段，统一成1,0,-1
     * ……
     */
    val beanRdd: RDD[AppLogBean] = filtered.
      rdd
      .map(BeanUtil.row2Bean(_))
      .map(bean => {
        val acc = bean.account
        if (StringUtils.isBlank(acc)) bean.account = null
        bean
      })

    // 分割session
    /**
     * 逻辑设计：
     * 先按原sessionid分组，将同一个session的所有行为事件放到一组（Iterator/List）
     * 然后对该组事件按时间先后顺序排序
     *
     * 构造一个新的sessionid
     * 然后遍历这个事件列表（循环）
     * 拿到一个事件bean，就给他赋值这个新sessionid
     * 如果下一个bean和当前bean的时间差>30分钟，则构造一个新的sessionid
     */
    val sessionSplitted: RDD[AppLogBean] = beanRdd
      // groupby后的结果是：(sessionid,Iterator[AppLogBean])
      .groupBy(b => b.sessionId)

      // mapvalues后得到的结果是：(sessionid,Iterator[AppLogBean])
      .mapValues(iter => {
        // 对同一个session的所有行为事件，按时间先后排序
        val orderedList = iter.toList.sortBy(bean => bean.timeStamp)

        // 生成一个新一套session的初始值
        var ns = UUID.randomUUID().toString
        for (i <- 0 until orderedList.size) {

          // 赋值新的一套sessionid
          orderedList(i).newsession = ns

          // 判断相邻两个事件之间的时间间隔，是否超出30分钟
          // 生成一个新的sessionid
          if (i < orderedList.size - 1 && orderedList(i + 1).timeStamp - orderedList(i).timeStamp > 30 * 60 * 1000) {
            ns = UUID.randomUUID().toString
          }
        }

        // 返回处理好的事件list
        orderedList
      })
      .flatMap(_._2)


    /**
     * gps坐标，匹配省市区地理位置信息
     */

    // 方法1： 利用redis的geo查询功能
    /*sessionSplitted.map(bean => {
      val latitude = bean.latitude
      val longitude = bean.longitude

      // 构造一个redis的连接
      val jedis = new Jedis("doitedu01", 6379)

      // 利用客户端去redis服务器中查找当前坐标点最近的样本点，得到其地理位置信息
      val responses: util.List[GeoRadiusResponse] = jedis.georadius("geodict", longitude, latitude, 3, GeoUnit.KM)

      if (responses != null && responses.size() > 0) {
        val areaInfo = responses.get(0).getMemberByString
        val strings = areaInfo.split(":")

        bean.province = strings(0)
        bean.city = strings(1)
        bean.district = strings(2)
      }

    })*/

    // 方法2： 利用geohash编码，自己来进行查询匹配
    // 将深似海男人曾经给的全国各区县样本点知识表（mysql表），加工成扁平结构 ：经度,维度,省,市,区
    // 然后将知识数据中的 经纬度，换算成geohash码：5位（base32编码后的位数）
    // 然后将这个编码后的坐标，和省市区信息，封装到一个HashMap中，并广播出去

    // 对日志处理，map(
    //   拿到一个bean，取出他的经纬度，并转成geohash编码（5位）
    //   然后拿这个编码，去广播变量（hashmap）中get地理位置信息
    // )

    val props = new Properties()
    props.setProperty("user", "root")
    props.setProperty("password", "ABC123abc.123")
    val area = spark.read.jdbc("jdbc:mysql://doitedu01:3306/realtimedw", "t_md_areas", props)
    area.createTempView("area")


    /**
     * 定义一个udf函数：输入经纬度，返回geohash编码
     */
    val geoHash = (lat: Double, lng: Double) => {
      GeoHash.geoHashStringWithCharacterPrecision(lat, lng, 5)
    }
    spark.udf.register("gps2geo", geoHash)


    val area2 = spark.sql(
      """
        |
        |
        |select
        |
        |gps2geo(d.bd09_lat,d.bd09_lng) as geohash,
        |p.areaname as province,
        |c.areaname as city,
        |d.areaname as district
        |
        |
        |from (select * from area where level=3) d join area c on d.parentid = c.id
        |            join area p on c.parentid = p.id
        |
        |""".stripMargin)

    import org.apache.spark.sql.functions._
    area2.filter(col("id") > 3)
    area2.filter(row=>row.getAs[Int]("id")>3)

    area2.filter(col("sessionid").isNotNull && trim(col("sessionid")) != lit(""))



    // 将area2（RDD）收集到driver端成为一个单机集合
    val geotemp = area2.rdd.map(row => {
      val geo = row.getAs[String]("geohash")
      val province = row.getAs[String]("province")
      val city = row.getAs[String]("city")
      val district = row.getAs[String]("district")
      (geo, (province, city, district))
    })

    //geotemp.saveAsTextFile("file:///d:/geo.txt")
    //sys.exit()

    val geoMap = geotemp.collectAsMap()
    // 广播
    val bc = spark.sparkContext.broadcast(geoMap)


    // 读取ip2region的ip地理位置知识库文件，成为一个字节数组，并广播
    val fs: FileSystem = FileSystem.get(new URI("hdfs://doitedu01:8020"), new Configuration(), "root")
    val in: FSDataInputStream = fs.open(new Path("/ip2region/ip2region.db"))

    val f = fs.listStatus(new Path("/ip2region/"))(0)

    val buffer = new Array[Byte](f.getLen.toInt)
    in.readFully(buffer)

    val bc_ip = spark.sparkContext.broadcast(buffer)


    /**
     * 利用gps坐标和ip地址，集成地理位置信息
     */
    val regioned: RDD[AppLogBean] = sessionSplitted.mapPartitions(iter => {

      // 取出geo地理位置广播变量
      val geoHashMap = bc.value
      // 取出ip地理位置广播变量
      val ipDbBytes = bc_ip.value

      iter.map(bean => {
        val lat = bean.latitude
        val lng = bean.longitude

        // 转成geohash编码
        var flag = false

        if (lat > -90 && lat < 90 && lng > -180 && lng < 180) {
          val logGeo = GeoHash.geoHashStringWithCharacterPrecision(lat, lng, 5)

          // 用日志中的gps geohash编码，去知识hashmap中取对应的省市区信息
          val maybeTuple: Option[(String, String, String)] = geoHashMap.get(logGeo)
          if (maybeTuple.isDefined) {
            val areaInfo = maybeTuple.get
            bean.province = areaInfo._1
            bean.city = areaInfo._2
            bean.district = areaInfo._3

            // 重置标记：地理位置查找成功
            flag = true
          }
        }


        // 如果前面步骤查找失败，则利用ip地址重新查找
        if (!flag) {
          val config = new DbConfig()
          val searcher = new DbSearcher(config, ipDbBytes)
          val block = searcher.memorySearch(bean.ip)
          // 中国|0|上海|上海市|电信
          val strings = block.getRegion.split("\\|")
          if (!strings(0).equals("0") && strings.size == 5) {
            bean.province = strings(2)
            bean.city = strings(3)
          }
        }

        bean
      }
      )
    })


    /**
     * 新老访客标记
     * 计算逻辑：
     *    1. 加载设备绑定评分表的T-1日分区
     *    2. 取出所有deviceid和account
     *    3. 组成一个set集合
     *    4. 广播
     *    5. 在日志中查询
     */
    val ids = spark.read
      .table("dws18.device_account_bind")
      .where(s"dt='${args(1)}'")
      .select("deviceid", "account")
      .rdd
      .flatMap(row => {
        val deviceid = row.getAs[String]("deviceid")
        val account = row.getAs[String]("account")
        Array(deviceid, account)
      }).collect().toSet.filter(_ != null)

    // val bc_ids = spark.sparkContext.broadcast(ids)


    /**
     * 构造一个存储所有ids的布隆过滤器
     */
    val filter = new BloomFilter(10000000, 5, Hash.JENKINS_HASH)
    for (elem <- ids) {
      filter.add(new Key(elem.getBytes()))
    }
    val bc_bloom = spark.sparkContext.broadcast(filter)


    val newUserFlaged = regioned.map(bean => {
      val bloomFilter = bc_bloom.value

      val did = bean.deviceId
      val acc = bean.account

      // 如果在历史id列表中包含本deviceid或本account
      var flag = true
      var isnew = true

      // 判断did是否存在
      if (bloomFilter.membershipTest(new Key(did.getBytes()))) {
        // 如果存在，则关闭标记，后面代码不执行，isnew=false
        flag = false
        isnew = false
      }

      // 如果did不存在，则继续判断account
      if (flag && acc != null) {
        if (bloomFilter.membershipTest(new Key(acc.getBytes()))) isnew = false
      }

      bean.isnew = if(isnew) 1 else 0
      bean
    })


    /**
     * idmapping
     *
     * 添加一个字段： guid
     *
     * 计算思想：
     *  1. 将日志数据 left join 绑定评分表，连接条件：  deviceid
     *  2. 如果日志中有account，则 guid= 日志表的account
     *  3. 如果日志中没有account，而评分表中有account，则guid = 评分表的account
     *  4. 如果日志和评分表中都没有account，则 guid = 设备id
     */

    spark.read
      .table("dws18.device_account_bind")
      .where(s"dt='${args(2)}'")
      .createTempView("bind")

    // 处理评分表，一个设备id，只保留分数最高的一个account
    spark.sql(
      """
        |select
        |deviceid,account
        |from (
        |   select
        |     deviceid,
        |     account,
        |     row_number() over(partition by deviceid order by score desc,timestamp desc) as rn
        |   from bind
        |) o
        |where rn=1
        |
        |""".stripMargin).createTempView("bind2")


    newUserFlaged.toDF().createTempView("regioned")


    /**
     * 添加guid字段
     *
     * 并输出最终结果  dwd层: dwd18.event_app_detail (dt = 2020-12-11)
     *
     */
    val guided = spark.sql(
      s"""
         |
         |INSERT INTO TABLE dwd18.event_app_detail PARTITION(dt='${args(2)}')
         |
         |select
         | regioned.account          ,
         | regioned.appid            ,
         | regioned.appversion       ,
         | regioned.carrier          ,
         | regioned.deviceid         ,
         | regioned.devicetype       ,
         | regioned.eventid          ,
         | regioned.ip               ,
         | regioned.latitude         ,
         | regioned.longitude        ,
         | regioned.nettype          ,
         | regioned.osname           ,
         | regioned.osversion        ,
         | regioned.properties       ,
         | regioned.releasechannel   ,
         | regioned.resolution       ,
         | regioned.sessionid        ,
         | regioned.timestamp        ,
         | regioned.newsession       ,
         | regioned.province   	    ,
         | regioned.city             ,
         | regioned.district         ,
         | regioned.isnew            ,
         | case
         |   when regioned.account is not null then regioned.account
         |   when bind2.account is not null then bind2.account
         |   else regioned.deviceid
         | end as guid
         |
         |from regioned left join bind2 on bind2.deviceid=regioned.deviceid
         |
         |""".stripMargin)


    spark.close()
  }
}
